package com.kuding.log.components;

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.InitializingBean;

import com.aliyun.openservices.loghub.client.config.LogHubConfig;
import com.aliyun.openservices.loghub.client.exceptions.LogHubClientWorkerException;
import com.aliyun.openservices.loghub.client.interfaces.ILogHubProcessorFactory;
import com.kuding.exceptions.VinusException;
import com.kuding.props.AliyunLogProperties;

public class LogWorkerContainer implements InitializingBean, DisposableBean {

	private final static Log logger = LogFactory.getLog(LogWorkerContainer.class);

	private final Map<String, VinusClientWorker> clientWorkers = new ConcurrentHashMap<>();

	private final AliyunLogProperties aliyunLogProperties;

	private final ExecutorService executorService;

	private final ILogHubProcessorFactory logHubProcessorFactory;

	public LogWorkerContainer(AliyunLogProperties aliyunLogProperties, ExecutorService executorService,
			ILogHubProcessorFactory logHubProcessorFactory) {
		this.aliyunLogProperties = aliyunLogProperties;
		this.executorService = executorService;
		this.logHubProcessorFactory = logHubProcessorFactory;
	}

	@Override
	public void afterPropertiesSet() throws Exception {
		logger.info("init client worker:" + aliyunLogProperties.getLogStores().keySet());
		aliyunLogProperties.getLogStores().forEach((x, y) -> {
			LogHubConfig logHubConfig = y.generateLogHubConfig(aliyunLogProperties, x);
			try {
				VinusClientWorker clientWorker = new VinusClientWorker(logHubProcessorFactory, logHubConfig,
						executorService, x);
				clientWorkers.put(x, clientWorker);
				executorService.execute(clientWorker);
			} catch (LogHubClientWorkerException e) {
				throw new VinusException("create client worker error! ", e);
			}
		});
	}

	@Override
	public void destroy() throws Exception {
		logger.info("shutdown client workers");
		clientWorkers.values().forEach(x -> x.shutdown());
	}

}
